using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Text;
using System.Threading;
using CoreLab.Oracle;
using Common.Utilities;
using Common.Utilities.Dispatcher;

namespace Workflows.Components.ETL
{
    /// <summary>
    /// sync table data between different database, can be oracle and sql server
    /// do not track data versions, only most recent data is kept
    /// </summary>
    public class SyncDbTableWorker : IWorker
    {
        #region IWorker Members
        private Thread _WorkerThread;
        private SyncDbTableInstruction _Instruction;
        private SyncDbTableOutput _Output;
        private Guid _TicketID;
        private RunContext _Context;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }

        #region const

        private const string TgtTableTimestampFieldName = "UpdatedOn";
        #endregion

        /// <summary>
        /// thread handle
        /// </summary>
        public Thread WorkerThread
        {
            get { return this._WorkerThread; }
        }

        /// <summary>
        /// instruction
        /// </summary>
        public InstructionBase Instruction
        {
            get { return this._Instruction; }
        }

        /// <summary>
        /// output: failed record IDs if there is a key field, otherwise, only count of failed updates
        /// </summary>
        public ResultBase Output
        {
            get { return this._Output; }
        }

        /// <summary>
        /// workflow component type associated with this worker
        /// </summary>
        public Type ComponentType
        {
            get { return typeof(SyncDbTableComponent); }
        }

        /// <summary>
        /// run context, app domain memory
        /// </summary>
        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }

        /// <summary>
        /// initialize worker
        /// </summary>
        /// <param name="ticketGuid"></param>
        /// <param name="request"></param>
        /// <param name="workflowName"></param>
        /// <param name="componentName"></param>
        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this._Instruction = (SyncDbTableInstruction) request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        /// <summary>
        /// initialize worker from within subflow
        /// </summary>
        /// <param name="context"></param>
        /// <param name="ticketGuid"></param>
        /// <param name="request"></param>
        /// <param name="workflowName"></param>
        /// <param name="componentName"></param>
        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._Context = context;
            this._TicketID = ticketGuid;
            this._Instruction = (SyncDbTableInstruction) request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        /// <summary>
        /// do work sync
        /// </summary>
        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = Thread.CurrentThread;
            this.SyncTable();
        }

        /// <summary>
        /// do work async
        /// </summary>
        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = new Thread(new ThreadStart(this.SyncTable));
            this._WorkerThread.Start();
        }

        /// <summary>
        /// cancel worker
        /// </summary>
        public void CancelWork()
        {
            if (this._WorkerThread != null && this._WorkerThread.IsAlive)
            {
                this._WorkerThread.Abort();

                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._TicketID, this.GetType(), this._Instruction);
                }
            }
        }

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;
        #endregion

        private string QuoteString1(DataProviderType dbProvider)
        {
            if (dbProvider == DataProviderType.Oracle)
                return "";
            else
                return "[";
        }
        private string QuoteString2(DataProviderType dbProvider)
        {
            if (dbProvider == DataProviderType.Oracle)
                return "";
            else
                return "]";
        }

        #region work
        private void SyncTable()
        {
            try
            {
                this.EnsureTgtTableSchema(this._Instruction.SrcDbProvider,
                    this._Instruction.SrcConnStr, this._Instruction.SrcSchema,
                    this._Instruction.SrcTblName, this._Instruction.SelectStatement,
                    this._Instruction.TgtDbProvider, this._Instruction.TgtConnStr,
                    this._Instruction.TgtTableName, this._Instruction.UseSql);

                int totalRecordsUpdated = 0;
                if (this._Instruction.SrcTableUniqueKeys == null || this._Instruction.SrcTableUniqueKeys.Count == 0)
                {
                    if (string.IsNullOrEmpty(this._Instruction.SrcTimeStampField))
                    {
                        totalRecordsUpdated = this.SyncTableInStreamingMode();
                    }
                    else
                    {
                        totalRecordsUpdated = this.SyncTableUsingTimestamp();
                    }
                }
                else
                {
                    totalRecordsUpdated = this.SyncTableUsingUniqueKeys();
                }
                this._Output = new SyncDbTableOutput(this._Instruction.OwnerComponentID, this._TicketID, totalRecordsUpdated);

                if (this.WorkerFinishedEvent != null)
                {
                    this.WorkerFinishedEvent(this._TicketID, this.GetType(), this._Instruction, this._Output);
                }
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
        }

        /// <summary>
        /// test method called sync by client
        /// </summary>
        /// <returns></returns>
        public DataTable TestTableSync()
        {
            DataTable dtTest = new DataTable();
            try
            {
                this.EnsureTgtTableSchema(this._Instruction.SrcDbProvider,
                    this._Instruction.SrcConnStr, this._Instruction.SrcSchema,
                    this._Instruction.SrcTblName, this._Instruction.SelectStatement,
                    this._Instruction.TgtDbProvider, this._Instruction.TgtConnStr,
                    this._Instruction.TgtTableName, this._Instruction.UseSql);

                if (this._Instruction.SrcTableUniqueKeys == null || this._Instruction.SrcTableUniqueKeys.Count == 0)
                {
                    if (string.IsNullOrEmpty(this._Instruction.SrcTimeStampField))
                    {
                        dtTest = this.TestTableSyncInStreamingMode();
                    }
                    else
                    {
                        dtTest = this.TestTableSyncUsingTimeStamp();
                    }
                }
                else
                {
                    dtTest = this.TestTableSyncUsingUniqueKeys();
                }
            }
            catch (Exception ex)
            {
                throw new Exception("Test failed: " + ex.ToString());
            }
            return dtTest;
        }
        #endregion

        #region sync table schema
        /// <summary>
        /// called by client and also within worker to ensure target table schema 
        /// can hold source data
        /// </summary>
        /// <param name="srcDbProvider"></param>
        /// <param name="srcConnStr"></param>
        /// <param name="srcSchema"></param>
        /// <param name="srcTblName"></param>
        /// <param name="selectSql"></param>
        /// <param name="tgtDbProvider"></param>
        /// <param name="tgtConnStr"></param>
        /// <param name="tgtTableName"></param>
        /// <param name="useSelectSql"></param>
        public void EnsureTgtTableSchema(
            DataProviderType srcDbProvider,
            string srcConnStr,
            string srcSchema,
            string srcTblName,
            string selectSql,
            DataProviderType tgtDbProvider,
            string tgtConnStr,
            string tgtTableName,
            bool useSelectSql)
        {
            TableColumn[] srcCols = null;
            if (useSelectSql)
            {
                srcCols =
                    DbUtil.GetTableColumns(srcDbProvider, srcConnStr, selectSql);
            }
            else
            {
                srcCols = DbUtil.GetTableColumns(srcDbProvider, srcConnStr, srcSchema, srcTblName);
            }
            List<TableColumn> tgtCols = new List<TableColumn>();
            if (this._Instruction.FieldMappings != null && this._Instruction.FieldMappings.Count > 0)
            {
                foreach(TableColumn srcCol in srcCols)
                {
                    if(this._Instruction.FieldMappings.ContainsKey(srcCol.ColumnName))
                    {
                        TableColumn tgtCol = srcCol;
                        tgtCol.ColumnName = this._Instruction.FieldMappings[srcCol.ColumnName];
                        tgtCols.Add(tgtCol);
                    }
                }
            }
            else
            {
                tgtCols.AddRange(srcCols);
            }
            //TableColumn updField = new TableColumn();
            //updField.AllowNull = false;
            //updField.ColumnName = TgtTableTimestampFieldName;
            //updField.DbType = SqlDbType.DateTime;
            //updField.IsFK = false;
            //updField.IsIdentity = false;
            //updField.IsIndexed = false;
            //updField.IsPK = false;
            //updField.Ordinal = srcCols.Length;
            //updField.Precision = 0;
            //updField.Scale = 0;
            //updField.Size = 8;
            //tgtCols.Add(updField);

            bool tableExist = DbUtil.IsTableExist(tgtDbProvider, tgtConnStr, tgtTableName);
            if (!tableExist)
            {
                try
                {
                    if(tgtDbProvider==DataProviderType.Oracle)
                    {
                        TableHelper.CreateOracleTable(tgtConnStr, tgtTableName, tgtCols.ToArray());
                    }
                    else
                    {
                        TableHelper.CreateTable(tgtConnStr, tgtTableName, tgtCols.ToArray(), true);    
                    }
                    
                }
                catch (Exception ex)
                {
                    throw new Exception("Unable to create target table: " + ex.ToString());
                }
            }
            else
            {
                TableColumn[] tgtCols2 =
                    DbUtil.GetTableColumns(tgtDbProvider, tgtConnStr, string.Empty, tgtTableName);
                int foundColumnCount = 0;
                foreach (TableColumn tgtCol in tgtCols)
                {
                    for (int i = 0; i < tgtCols2.Length; i++)
                    {
                        if (tgtCols2[i].ColumnName.ToLower() == tgtCol.ColumnName.ToLower())
                        {
                            foundColumnCount++;
                            break;
                        }
                    }
                }
                if (tgtCols.Count > foundColumnCount)
                {
                    throw new Exception("Invalid table schema in target database");
                }
            }
        }
        #endregion

        #region use streaming mode
        private int SyncTableInStreamingMode()
        {
            int recordsInSrcTable = 0;
            if (string.IsNullOrEmpty(this._Instruction.SelectStatement))
            {
                recordsInSrcTable =
                    DbUtil.GetTableRecordCount(
                    this._Instruction.SrcDbProvider,
                    this._Instruction.SrcConnStr, this._Instruction.SrcSchema,
                    this._Instruction.SrcTblName);
            }
            else
            {
                recordsInSrcTable =
                    DbUtil.GetTableRecordCount(
                        this._Instruction.SrcDbProvider,
                        this._Instruction.SrcConnStr,
                        this._Instruction.SelectStatement);
            }
            if (recordsInSrcTable == 0)
                return 0;

            int recordsInserted = 0;
            DateTime currentTimestamp = DateTime.Now;

            IDbConnection tgtConn;
            if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
            {
                tgtConn = new OracleConnection(this._Instruction.TgtConnStr);
            }
            else
            {
                tgtConn = new SqlConnection(this._Instruction.TgtConnStr);
            }
            IDbConnection srcConn;
            if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
            {
                srcConn = new OracleConnection(this._Instruction.SrcConnStr);
            }
            else
            {
                srcConn = new SqlConnection(this._Instruction.SrcConnStr);
            }
            try
            {
                TableColumn[] srcCols = null;
                if (this._Instruction.UseSql)
                {
                    srcCols =
                        DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                               this._Instruction.SelectStatement);
                }
                else
                {
                    srcCols = DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                                     this._Instruction.SrcSchema, this._Instruction.SrcTblName);
                }
                List<TableColumn> tgtCols = new List<TableColumn>();
                if (this._Instruction.FieldMappings != null && this._Instruction.FieldMappings.Count > 0)
                {
                    foreach (TableColumn srcCol in srcCols)
                    {
                        if (this._Instruction.FieldMappings.ContainsKey(srcCol.ColumnName))
                        {
                            TableColumn tgtCol = srcCol;
                            tgtCol.ColumnName = this._Instruction.FieldMappings[srcCol.ColumnName];
                            tgtCols.Add(tgtCol);
                        }
                    }
                }
                else
                {
                    tgtCols.AddRange(srcCols);
                }
                //TableColumn updField = new TableColumn();
                //updField.AllowNull = false;
                //updField.ColumnName = TgtTableTimestampFieldName;
                //updField.DbType = SqlDbType.DateTime;
                //updField.IsFK = false;
                //updField.IsIdentity = false;
                //updField.IsIndexed = false;
                //updField.IsPK = false;
                //updField.Ordinal = srcCols.Length;
                //updField.Precision = 0;
                //updField.Scale = 0;
                //updField.Size = 8;
                //tgtCols.Add(updField);

                srcConn.Open();
                tgtConn.Open();
                IDbCommand tgtCmd = tgtConn.CreateCommand();
                tgtCmd.CommandType = CommandType.Text;

                // determine identity insert, if true, set Identity Insert to Off
                bool identityInsert = false;
                foreach (TableColumn tgtCol in tgtCols)
                {
                    if (tgtCol.IsIdentity)
                    {
                        identityInsert = true;
                        break;
                    }
                }
                if (identityInsert)
                {
                    try
                    {
                        tgtCmd.CommandText = string.Format(
                            "SET IDENTITY_INSERT {0} ON;", this._Instruction.TgtTableName);
                        tgtCmd.ExecuteNonQuery();
                    }
                    catch
                    {
                        identityInsert = false;
                    }
                }


                string spName = "BulkInsert_" + this._Instruction.TgtTableName;
                if(this._Instruction.TgtDbProvider==DataProviderType.MSSQL)
                {
                    string dropSp = StoredProcUtil.CreateDropStoredProcedureStatement(this._Instruction.TgtDbProvider, spName);
                    tgtCmd.CommandText = dropSp;
                    tgtCmd.ExecuteNonQuery();
                }
                
                string createSp =
                    StoredProcUtil.CreateStoredProcedureStatementForInsert(this._Instruction.TgtDbProvider, spName,
                                                                           string.Empty, this._Instruction.TgtTableName,
                                                                           tgtCols.ToArray());
                tgtCmd.CommandText = createSp;
                tgtCmd.ExecuteNonQuery();

                tgtCmd.CommandText = "truncate table " + QuoteString1(this._Instruction.TgtDbProvider) +
                    this._Instruction.TgtTableName + QuoteString2(this._Instruction.TgtDbProvider);
                tgtCmd.ExecuteNonQuery();

                Dictionary<string, IDataParameter> parameters = new Dictionary<string, IDataParameter>();
                foreach (TableColumn tgtCol in tgtCols)
                {
                    if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
                    {
                        if (tgtCol.ColumnName == TgtTableTimestampFieldName)
                        {
                            IDataParameter parameter = new OracleParameter("@" + tgtCol.ColumnName, currentTimestamp);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                        else
                        {
                            IDataParameter parameter = new OracleParameter("@" + tgtCol.ColumnName, DBNull.Value);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                    }
                    else
                    {
                        if (tgtCol.ColumnName == TgtTableTimestampFieldName)
                        {
                            IDataParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, currentTimestamp);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                        else
                        {
                            IDataParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, DBNull.Value);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                    }
                }

                IDbCommand srcCmd = srcConn.CreateCommand();
                if (this._Instruction.UseSql)
                {
                    srcCmd.CommandText = this._Instruction.SelectStatement;
                }
                else
                {
                    string selectSql = "select *";
                    selectSql += " from " +
                                 (string.IsNullOrEmpty(this._Instruction.SrcSchema)
                                      ? QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider)
                                      : this._Instruction.SrcSchema + "." + QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider));
                    srcCmd.CommandText = selectSql;
                }

                srcCmd.CommandType = CommandType.Text;
                IDataReader srcReader = srcCmd.ExecuteReader();
                while (srcReader.Read())
                {
                    try
                    {
                        tgtCmd.Parameters.Clear();
                        // tgtCmd.Parameters.Add(parameters[TgtTableTimestampFieldName]);
                        foreach (TableColumn srcCol in srcCols)
                        {
                            string tgtColName = srcCol.ColumnName;
                            if (this._Instruction.FieldMappings != null &&
                                this._Instruction.FieldMappings.ContainsKey(srcCol.ColumnName))
                            {
                                tgtColName = this._Instruction.FieldMappings[srcCol.ColumnName];
                            }
                            if (parameters.ContainsKey(tgtColName))
                            {
                                IDataParameter parameter = parameters[tgtColName];
                                if (srcReader[srcCol.ColumnName] != null)
                                {
                                    object srcFieldValue = srcReader[srcCol.ColumnName];
                                    parameter.Value = srcFieldValue;
                                }
                                else
                                {
                                    parameter.Value = DBNull.Value;
                                }
                                tgtCmd.Parameters.Add(parameter);
                            }
                        }
                        tgtCmd.CommandText = spName;
                        tgtCmd.CommandType = CommandType.StoredProcedure;
                        tgtCmd.ExecuteNonQuery();
                    }
                    catch (Exception ex1)
                    {
                        if (this.WorkerErredEvent != null)
                        {
                            this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex1.ToString(),false);
                        }
                    }
                    recordsInserted++;
                    if (this.WorkerStatusChangedEvent != null && recordsInserted % 10 == 0)
                    {
                        string msg = string.Format("Adding records to table {0}, processed {1} of {2} records",
                                                   this._Instruction.TgtTableName, recordsInserted, recordsInSrcTable);
                        int percent = recordsInserted*100/recordsInSrcTable;
                        this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, percent);
                    }
                }
                srcReader.Close();

                if (identityInsert)
                {
                    tgtCmd.CommandType = CommandType.Text;
                    tgtCmd.CommandText = string.Format("SET IDENTITY_INSERT {0} OFF;", this._Instruction.TgtTableName);
                    tgtCmd.ExecuteNonQuery();
                }
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                srcConn.Close();
                tgtConn.Close();
            }

            return recordsInserted;
        }

        private DataTable TestTableSyncInStreamingMode()
        {
            DataTable testTable = new DataTable();
            IDbConnection srcConn;
            if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
            {
                srcConn = new OracleConnection(this._Instruction.SrcConnStr);
            }
            else
            {
                srcConn = new SqlConnection(this._Instruction.SrcConnStr);
            }
            try
            {
                TableColumn[] srcCols = null;
                if (this._Instruction.UseSql)
                {
                    srcCols =
                        DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                               this._Instruction.SelectStatement);
                }
                else
                {
                    srcCols = DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                                     this._Instruction.SrcSchema, this._Instruction.SrcTblName);
                }
                foreach (TableColumn col in srcCols)
                {
                    testTable.Columns.Add(col.ColumnName, FieldDataType.ToSystemType(col.DbType));
                }

                srcConn.Open();

                IDbCommand srcCmd = srcConn.CreateCommand();
                if (this._Instruction.UseSql)
                {
                    srcCmd.CommandText = this._Instruction.SelectStatement;
                }
                else
                {
                    string selectSql = "select *";
                    selectSql += " from " +
                                 (string.IsNullOrEmpty(this._Instruction.SrcSchema)
                                      ? QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider)
                                      : this._Instruction.SrcSchema + "." + QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider));
                    srcCmd.CommandText = selectSql;
                }

                srcCmd.CommandType = CommandType.Text;
                IDataReader srcReader = srcCmd.ExecuteReader();
                int recordsRead = 0;
                while (srcReader.Read())
                {
                    DataRow drTest = testTable.NewRow();
                    foreach (TableColumn srcCol in srcCols)
                    {
                        object srcFieldValue = srcReader[srcCol.ColumnName];
                        drTest[srcCol.ColumnName] = srcFieldValue;
                    }
                    testTable.Rows.Add(drTest);
                    recordsRead++;
                    if (recordsRead > 5)
                        break;
                }
                srcReader.Close();
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                srcConn.Close();
            }

            return testTable;
        }
        #endregion

        #region use timestamp
        private int SyncTableUsingTimestamp()
        {
            int recordsInSrcTable = 0;
            if (string.IsNullOrEmpty(this._Instruction.SelectStatement))
            {
                recordsInSrcTable =
                    DbUtil.GetTableRecordCount(
                    this._Instruction.SrcDbProvider,
                    this._Instruction.SrcConnStr, this._Instruction.SrcSchema,
                    this._Instruction.SrcTblName);
            }
            else
            {
                recordsInSrcTable =
                    DbUtil.GetTableRecordCount(
                        this._Instruction.SrcDbProvider,
                        this._Instruction.SrcConnStr,
                        this._Instruction.SelectStatement);
            }
            if (recordsInSrcTable == 0)
                return 0;

            int recordsInserted = 0;

            DateTime currentTimestamp = DateTime.Now;
            DateTime lastUpdateTimestamp = TimeUtil.DbMinTime;
            IDbConnection tgtConn;
            if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
            {
                tgtConn = new OracleConnection(this._Instruction.TgtConnStr);
            }
            else
            {
                tgtConn = new SqlConnection(this._Instruction.TgtConnStr);
            }
            IDbConnection srcConn;
            if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
            {
                srcConn = new OracleConnection(this._Instruction.SrcConnStr);
            }
            else
            {
                srcConn = new SqlConnection(this._Instruction.SrcConnStr);
            }
            try
            {
                TableColumn[] srcCols = null;
                if (this._Instruction.UseSql)
                {
                    srcCols =
                        DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                               this._Instruction.SelectStatement);
                }
                else
                {
                    srcCols = DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                                     this._Instruction.SrcSchema, this._Instruction.SrcTblName);
                }
                List<TableColumn> tgtCols = new List<TableColumn>();
                if (this._Instruction.FieldMappings != null && this._Instruction.FieldMappings.Count > 0)
                {
                    foreach (TableColumn srcCol in srcCols)
                    {
                        if (this._Instruction.FieldMappings.ContainsKey(srcCol.ColumnName))
                        {
                            TableColumn tgtCol = srcCol;
                            tgtCol.ColumnName = this._Instruction.FieldMappings[srcCol.ColumnName];
                            tgtCols.Add(tgtCol);
                        }
                    }
                }
                else
                {
                    tgtCols.AddRange(srcCols);
                }
                TableColumn updField = new TableColumn();
                updField.AllowNull = false;
                updField.ColumnName = TgtTableTimestampFieldName;
                updField.DbType = SqlDbType.DateTime;
                updField.IsFK = false;
                updField.IsIdentity = false;
                updField.IsIndexed = false;
                updField.IsPK = false;
                updField.Ordinal = srcCols.Length;
                updField.Precision = 0;
                updField.Scale = 0;
                updField.Size = 8;
                tgtCols.Add(updField);

                srcConn.Open();
                tgtConn.Open();
                IDbCommand tgtCmd = tgtConn.CreateCommand();
                tgtCmd.CommandText = "select max (" + TgtTableTimestampFieldName + ") from " +
                                     this._Instruction.TgtTableName;
                tgtCmd.CommandType = CommandType.Text;
                // determine identity insert, if true, set Identity Insert to Off
                bool identityInsert = false;
                foreach (TableColumn tgtCol in tgtCols)
                {
                    if (tgtCol.IsIdentity)
                    {
                        identityInsert = true;
                        break;
                    }
                }
                if (identityInsert)
                {
                    tgtCmd.CommandText = string.Format("SET IDENTITY_INSERT {0} ON;", this._Instruction.TgtTableName);
                    tgtCmd.ExecuteNonQuery();
                }

                IDataReader tgtReader = tgtCmd.ExecuteReader();
                if (tgtReader.Read())
                {
                    lastUpdateTimestamp = tgtReader.GetDateTime(0);
                }
                tgtReader.Close();

                string spName = "BulkInsert_" + this._Instruction.TgtTableName;
                if (this._Instruction.TgtDbProvider == DataProviderType.MSSQL)
                {
                    string dropSp =
                        StoredProcUtil.CreateDropStoredProcedureStatement(this._Instruction.TgtDbProvider, spName);
                    tgtCmd.CommandText = dropSp;
                    tgtCmd.CommandType = CommandType.Text;
                    tgtCmd.ExecuteNonQuery();
                }
                string createSp =
                    StoredProcUtil.CreateStoredProcedureStatementForInsert(this._Instruction.TgtDbProvider, spName,
                                                                           string.Empty, this._Instruction.TgtTableName,
                                                                           tgtCols.ToArray());
                tgtCmd.CommandText = createSp;
                tgtCmd.ExecuteNonQuery();

                Dictionary<string, IDataParameter> parameters = new Dictionary<string, IDataParameter>();
                foreach (TableColumn tgtCol in tgtCols)
                {
                    if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
                    {
                        if (tgtCol.ColumnName == TgtTableTimestampFieldName)
                        {
                            IDataParameter parameter = new OracleParameter("@" + tgtCol.ColumnName, currentTimestamp);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                        else
                        {
                            IDataParameter parameter = new OracleParameter("@" + tgtCol.ColumnName, DBNull.Value);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                    }
                    else
                    {
                        if (tgtCol.ColumnName == TgtTableTimestampFieldName)
                        {
                            IDataParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, currentTimestamp);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                        else
                        {
                            IDataParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, DBNull.Value);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                    }
                }

                IDbCommand srcCmd = srcConn.CreateCommand();
                string selectSql = string.Empty;
                if (this._Instruction.UseSql)
                {
                    selectSql = this._Instruction.SelectStatement;
                }
                else
                {
                    selectSql = "select * from " +
                                (string.IsNullOrEmpty(this._Instruction.SrcSchema)
                                     ? QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider)
                                     : this._Instruction.SrcSchema + "." + QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider));
                }
                string whereSql = string.Empty;


                if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
                {
                    whereSql = " where " + this._Instruction.SrcTimeStampField + ">" +
                               FieldDataType.ToOracleDataValue(SqlDbType.DateTime,
                                                               lastUpdateTimestamp.ToString()) + " and " +
                               this._Instruction.SrcTimeStampField + "<" +
                               FieldDataType.ToOracleDataValue(SqlDbType.DateTime, currentTimestamp.ToString());
                }
                else
                {
                    whereSql = " where " + this._Instruction.SrcTimeStampField + ">" +
                               FieldDataType.ToMssqlDataValue(SqlDbType.DateTime,
                                                              lastUpdateTimestamp.ToString()) + " and " +
                               this._Instruction.SrcTimeStampField + "<" +
                               FieldDataType.ToMssqlDataValue(SqlDbType.DateTime, currentTimestamp.ToString());
                }
                if (selectSql.ToLower().IndexOf("where") > 0)
                {
                    whereSql = " and " + whereSql.Substring(whereSql.ToLower().IndexOf("where") + "where".Length);
                }
                srcCmd.CommandText = selectSql + whereSql;

                IDataReader srcReader = srcCmd.ExecuteReader();

                tgtCmd.CommandText = spName;
                tgtCmd.CommandType = CommandType.StoredProcedure;

                while (srcReader.Read())
                {
                    try
                    {
                        tgtCmd.Parameters.Clear();
                        tgtCmd.Parameters.Add(parameters[TgtTableTimestampFieldName]);
                        foreach (TableColumn srcCol in srcCols)
                        {
                            string tgtColName = srcCol.ColumnName;
                            if(this._Instruction.FieldMappings !=null && this._Instruction.FieldMappings.ContainsKey(srcCol.ColumnName))
                            {
                                tgtColName = this._Instruction.FieldMappings[srcCol.ColumnName];
                            }
                            if(parameters.ContainsKey(tgtColName))
                            {
                                IDataParameter parameter = parameters[tgtColName];
                                if(srcReader[srcCol.ColumnName]!=null)
                                {
                                    object srcFieldValue = srcReader[srcCol.ColumnName];
                                    parameter.Value = srcFieldValue;
                                }
                                else
                                    parameter.Value = DBNull.Value;
                                tgtCmd.Parameters.Add(parameter);
                            }
                        }
                        tgtCmd.ExecuteNonQuery();
                    }
                    catch (Exception ex1)
                    {
                        if (this.WorkerErredEvent != null)
                        {
                            this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex1.ToString(),false);
                        }
                    }
                    recordsInserted++;

                    if (this.WorkerStatusChangedEvent != null && recordsInserted % 10 == 0)
                    {
                        string msg =
                            string.Format("adding records to table {0}: processed {1} of {2} records.",
                                          this._Instruction.TgtTableName,
                                          recordsInserted,
                                          recordsInSrcTable);
                        int percent = recordsInserted*100/recordsInSrcTable;
                        this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, percent);
                    }
                }
                srcReader.Close();

                if (identityInsert)
                {
                    tgtCmd.CommandType = CommandType.Text;
                    tgtCmd.CommandText = string.Format("SET IDENTITY_INSERT {0} OFF;", this._Instruction.TgtTableName);
                    tgtCmd.ExecuteNonQuery();
                }
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                srcConn.Close();
                tgtConn.Close();
            }

            return recordsInserted;
        }

        private DataTable TestTableSyncUsingTimeStamp()
        {
            DataTable testTable = new DataTable();

            DateTime currentTimestamp = DateTime.Now;
            DateTime lastUpdateTimestamp = TimeUtil.DbMinTime;
            IDbConnection srcConn;
            if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
            {
                srcConn = new OracleConnection(this._Instruction.SrcConnStr);
            }
            else
            {
                srcConn = new SqlConnection(this._Instruction.SrcConnStr);
            }
            try
            {
                TableColumn[] srcCols = null;
                if (this._Instruction.UseSql)
                {
                    srcCols =
                        DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                               this._Instruction.SelectStatement);
                }
                else
                {
                    srcCols = DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                                     this._Instruction.SrcSchema, this._Instruction.SrcTblName);
                }
                foreach (TableColumn col in srcCols)
                {
                    testTable.Columns.Add(col.ColumnName, FieldDataType.ToSystemType(col.DbType));
                }

                srcConn.Open();
                IDbCommand srcCmd = srcConn.CreateCommand();
                string selectSql = string.Empty;
                if (this._Instruction.UseSql)
                {
                    selectSql = this._Instruction.SelectStatement;
                }
                else
                {
                    selectSql = "select * from " +
                                (string.IsNullOrEmpty(this._Instruction.SrcSchema)
                                     ? QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider)
                                     : this._Instruction.SrcSchema + "." + QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider));
                }
                string whereSql = string.Empty;


                if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
                {
                    whereSql = " where " + this._Instruction.SrcTimeStampField + ">" +
                               FieldDataType.ToOracleDataValue(SqlDbType.DateTime,
                                                               lastUpdateTimestamp.ToString()) + " and " +
                               this._Instruction.SrcTimeStampField + "<" +
                               FieldDataType.ToOracleDataValue(SqlDbType.DateTime, currentTimestamp.ToString());
                }
                else
                {
                    whereSql = " where " + this._Instruction.SrcTimeStampField + ">" +
                               FieldDataType.ToMssqlDataValue(SqlDbType.DateTime,
                                                              lastUpdateTimestamp.ToString()) + " and " +
                               this._Instruction.SrcTimeStampField + "<" +
                               FieldDataType.ToMssqlDataValue(SqlDbType.DateTime, currentTimestamp.ToString());
                }
                if (selectSql.ToLower().IndexOf("where") > 0)
                {
                    whereSql = " and " + whereSql.Substring(whereSql.ToLower().IndexOf("where") + "where".Length);
                }
                srcCmd.CommandText = selectSql + whereSql;
                IDataReader srcReader = srcCmd.ExecuteReader();
                int recordTested = 0;
                while (srcReader.Read())
                {
                    DataRow drTest = testTable.NewRow();
                    foreach (TableColumn srcCol in srcCols)
                    {
                        object srcFieldValue = srcReader[srcCol.ColumnName];
                        drTest[srcCol.ColumnName] = srcFieldValue;
                    }
                    testTable.Rows.Add(drTest);

                    recordTested++;
                    if (recordTested > 5)
                        break;
                }
                srcReader.Close();
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                srcConn.Close();
            }
            return testTable;
        }
        #endregion

        #region use unique field(s)
        private int SyncTableUsingUniqueKeys()
        {
            int recordsInTgtTable =
                DbUtil.GetTableRecordCount(
                    this._Instruction.TgtDbProvider,
                    this._Instruction.TgtConnStr,
                    string.Empty, this._Instruction.TgtTableName);
            if (recordsInTgtTable == 0)
                return this.SyncTableInStreamingMode();

            int recordsInSrcTable = 0;
            if (string.IsNullOrEmpty(this._Instruction.SelectStatement))
            {
                recordsInSrcTable =
                    DbUtil.GetTableRecordCount(
                    this._Instruction.SrcDbProvider,
                    this._Instruction.SrcConnStr, this._Instruction.SrcSchema,
                    this._Instruction.SrcTblName);
            }
            else
            {
                recordsInSrcTable =
                    DbUtil.GetTableRecordCount(
                        this._Instruction.SrcDbProvider,
                        this._Instruction.SrcConnStr,
                        this._Instruction.SelectStatement);
            }
            if (recordsInSrcTable == 0)
                return 0;

            int recordsInserted = 0;
            DateTime currentTimestamp = DateTime.Now;

            IDbConnection tgtConn;
            if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
            {
                tgtConn = new OracleConnection(this._Instruction.TgtConnStr);
            }
            else
            {
                tgtConn = new SqlConnection(this._Instruction.TgtConnStr);
            }
            IDbConnection srcConn;
            if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
            {
                srcConn = new OracleConnection(this._Instruction.SrcConnStr);
            }
            else
            {
                srcConn = new SqlConnection(this._Instruction.SrcConnStr);
            }
            try
            {
                TableColumn[] srcCols = null;
                if (this._Instruction.UseSql)
                {
                    srcCols =
                        DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                               this._Instruction.SelectStatement);
                }
                else
                {
                    srcCols = DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                                     this._Instruction.SrcSchema, this._Instruction.SrcTblName);
                }

                List<TableColumn> tgtCols = new List<TableColumn>();
                if (this._Instruction.FieldMappings != null && this._Instruction.FieldMappings.Count > 0)
                {
                    foreach (TableColumn srcCol in srcCols)
                    {
                        if (this._Instruction.FieldMappings.ContainsKey(srcCol.ColumnName))
                        {
                            TableColumn tgtCol = srcCol;
                            tgtCol.ColumnName = this._Instruction.FieldMappings[srcCol.ColumnName];
                            tgtCols.Add(tgtCol);
                        }
                    }
                }
                else
                {
                    tgtCols.AddRange(srcCols);
                }
                //TableColumn updField = new TableColumn();
                //updField.AllowNull = false;
                //updField.ColumnName = TgtTableTimestampFieldName;
                //updField.DbType = SqlDbType.DateTime;
                //updField.IsFK = false;
                //updField.IsIdentity = false;
                //updField.IsIndexed = false;
                //updField.IsPK = false;
                //updField.Ordinal = srcCols.Length;
                //updField.Precision = 0;
                //updField.Scale = 0;
                //updField.Size = 8;
                //tgtCols.Add(updField);

                srcConn.Open();
                tgtConn.Open();
                IDbCommand tgtCmd = tgtConn.CreateCommand();
                // determine identity insert, if true, set Identity Insert to Off
                bool identityInsert = false;
                foreach (TableColumn tgtCol in tgtCols)
                {
                    if (tgtCol.IsIdentity)
                    {
                        identityInsert = true;
                        break;
                    }
                }
                if (identityInsert)
                {
                    tgtCmd.CommandText = string.Format("SET IDENTITY_INSERT {0} ON;", this._Instruction.TgtTableName);
                    tgtCmd.ExecuteNonQuery();
                }

                string spName = "BulkInsert_" + this._Instruction.TgtTableName;
                if (this._Instruction.TgtDbProvider == DataProviderType.MSSQL)
                {
                    string dropSp =
                        StoredProcUtil.CreateDropStoredProcedureStatement(this._Instruction.TgtDbProvider, spName);
                    tgtCmd.CommandText = dropSp;
                    tgtCmd.CommandType = CommandType.Text;
                    tgtCmd.ExecuteNonQuery();
                }
                string createSp =
                    StoredProcUtil.CreateStoredProcedureStatementForInsert(this._Instruction.TgtDbProvider, spName,
                                                                           string.Empty, this._Instruction.TgtTableName,
                                                                           tgtCols.ToArray());
                tgtCmd.CommandText = createSp;
                tgtCmd.ExecuteNonQuery();

                Dictionary<string, IDataParameter> parameters = new Dictionary<string, IDataParameter>();
                foreach (TableColumn tgtCol in tgtCols)
                {
                    if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
                    {
                        if (tgtCol.ColumnName == TgtTableTimestampFieldName)
                        {
                            IDataParameter parameter = new OracleParameter("@" + tgtCol.ColumnName, currentTimestamp);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                        else
                        {
                            IDataParameter parameter = new OracleParameter("@" + tgtCol.ColumnName, DBNull.Value);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                    }
                    else
                    {
                        if (tgtCol.ColumnName == TgtTableTimestampFieldName)
                        {
                            IDataParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, currentTimestamp);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                        else
                        {
                            IDataParameter parameter = new SqlParameter("@" + tgtCol.ColumnName, DBNull.Value);
                            parameters.Add(tgtCol.ColumnName, parameter);
                        }
                    }
                }

                SqlDbType keyFieldDataType = SqlDbType.VarChar;
                string uniqueCompositeKeyName = this._Instruction.SrcTableUniqueKeys[0];
                string uniqueCompositeKeyExp = this._Instruction.SrcTableUniqueKeys[0];
                if (this._Instruction.SrcTableUniqueKeys.Count > 1)
                {
                    uniqueCompositeKeyExp = this.GetConvertionSqlExpression(this._Instruction.SrcDbProvider,
                                                                            this._Instruction.SrcTableUniqueKeys[0]);
                    for (int i = 1; i < this._Instruction.SrcTableUniqueKeys.Count;i++)
                    {
                        uniqueCompositeKeyExp += "+'-'+" +
                                                 this.GetConvertionSqlExpression(this._Instruction.SrcDbProvider,
                                                                                 this._Instruction.SrcTableUniqueKeys[i]);
                    }
                    uniqueCompositeKeyName = "CompositeUniqueKey";
                }
                else
                {
                    foreach(TableColumn srcCol in srcCols)
                    {
                        if(srcCol.ColumnName.ToUpper()==this._Instruction.SrcTableUniqueKeys[0].ToUpper())
                        {
                            keyFieldDataType = srcCol.DbType;
                            break;
                        }
                    }
                }
                IDbCommand srcCmd = srcConn.CreateCommand();
                // collect unique key values from target table 
                tgtCmd.CommandType = CommandType.Text;
                tgtCmd.CommandText = "select " + uniqueCompositeKeyExp +
                                     " from " + this._Instruction.TgtTableName;
                List<string> uniqueKeysInTgtTable = new List<string>();
                List<string> uniqueKeysInSrcTable = new List<string>();
                IDataReader tgtReader = tgtCmd.ExecuteReader();
                while(tgtReader.Read())
                {
                    if (tgtReader[0] != null && tgtReader[0] != DBNull.Value)
                    {
                        string uniqueKeyInTgt = tgtReader[0].ToString();
                        if (!string.IsNullOrEmpty(uniqueKeyInTgt))
                        {
                            uniqueKeyInTgt = uniqueKeyInTgt.ToUpper();
                            if (!uniqueKeysInTgtTable.Contains(uniqueKeyInTgt))
                            {
                                uniqueKeysInTgtTable.Add(uniqueKeyInTgt);
                            }
                        }
                    }
                }
                tgtReader.Close();

                if (this._Instruction.UseSql)
                {
                    string selectSql =
                        this._Instruction.SelectStatement.Substring(0,
                                                                    this._Instruction.SelectStatement.ToLower().IndexOf(
                                                                        "from"));
                    string fromSql =
                        this._Instruction.SelectStatement.Substring(
                            this._Instruction.SelectStatement.ToLower().IndexOf("from"));

                    if (this._Instruction.SrcTableUniqueKeys.Count > 1)
                    {
                        selectSql += "," + uniqueCompositeKeyExp + " as " + uniqueCompositeKeyName;
                    }
                    selectSql += fromSql;
                    srcCmd.CommandText = selectSql;
                }
                else
                {
                    string selectSql = "select *";
                    if (this._Instruction.SrcTableUniqueKeys.Count > 1)
                    {
                        selectSql += "," + uniqueCompositeKeyExp + " as " + uniqueCompositeKeyName;
                    }
                    selectSql += " from " +
                                 (string.IsNullOrEmpty(this._Instruction.SrcSchema)
                                      ? QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider)
                                      : this._Instruction.SrcSchema + "." + QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider));
                    srcCmd.CommandText = selectSql;
                }

                srcCmd.CommandType = CommandType.Text;
                IDataReader srcReader = srcCmd.ExecuteReader();
                
                while (srcReader.Read())
                {
                    if(srcReader[0]==null || srcReader[0]==DBNull.Value)
                        continue;

                    string uniqueKeyValue = srcReader[uniqueCompositeKeyName].ToString().ToUpper();
                    if (!uniqueKeysInSrcTable.Contains(uniqueKeyValue))
                        uniqueKeysInSrcTable.Add(uniqueKeyValue);
                    bool exist = false;
                    if(!string.IsNullOrEmpty(uniqueKeyValue) &&
                        uniqueKeysInTgtTable.Contains(uniqueKeyValue))
                    {
                        exist = true;
                    }
                    
                    if (!exist)
                    {
                        try
                        {
                            tgtCmd.Parameters.Clear();
                            // tgtCmd.Parameters.Add(parameters[TgtTableTimestampFieldName]);
                            foreach (TableColumn srcCol in srcCols)
                            {
                                string tgtColName = srcCol.ColumnName;
                                if(this._Instruction.FieldMappings !=null && this._Instruction.FieldMappings.ContainsKey(srcCol.ColumnName))
                                {
                                    tgtColName = this._Instruction.FieldMappings[srcCol.ColumnName];
                                }
                                if (parameters.ContainsKey(tgtColName))
                                {
                                    IDataParameter parameter = parameters[tgtColName];
                                    if (srcReader[srcCol.ColumnName] != null)
                                        parameter.Value = srcReader[srcCol.ColumnName];
                                    else
                                        parameter.Value = DBNull.Value;
                                    tgtCmd.Parameters.Add(parameter);
                                }
                            }
                            tgtCmd.CommandText = spName;
                            tgtCmd.CommandType = CommandType.StoredProcedure;
                            tgtCmd.ExecuteNonQuery();
                            uniqueKeysInTgtTable.Add(uniqueKeyValue.ToUpper());
                        }
                        catch (Exception ex1)
                        {
                            if (this.WorkerErredEvent != null)
                            {
                                this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex1.ToString(),false);
                            }
                        }

                        recordsInserted++;
                        if (this.WorkerStatusChangedEvent != null && recordsInserted % 100 == 0)
                        {
                            string msg = string.Format("Adding new records to table {0}, processed {1} of {2} records",
                                                       this._Instruction.TgtTableName, recordsInserted,
                                                       recordsInSrcTable);
                            int percent = recordsInserted * 100 / recordsInSrcTable;
                            this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this._Instruction,
                                this.WorkflowName, this.ComponentName, msg, percent);
                        }
                    }
                }
                srcReader.Close();

                if(this.WorkerStatusChangedEvent !=null)
                {
                    this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                  "Determining records needs to be deleted in table " +
                                                  this._Instruction.TgtTableName +
                                                  "...", 96);
                }
                tgtCmd.CommandText = "select " + uniqueCompositeKeyExp + " as " +
                                     uniqueCompositeKeyName + " from " + QuoteString1(this._Instruction.TgtDbProvider) +
                                     this._Instruction.TgtTableName + QuoteString2(this._Instruction.TgtDbProvider);
                tgtCmd.CommandType = CommandType.Text;
                List<string> keysToDelete = new List<string>();
                tgtReader = tgtCmd.ExecuteReader();
                while (tgtReader.Read())
                {
                    if(tgtReader[0]==null || tgtReader[0]==DBNull.Value)
                        continue;

                    string keyValue = tgtReader[0].ToString();
                    if(!string.IsNullOrEmpty(keyValue) && 
                        !uniqueKeysInSrcTable.Contains(keyValue.ToUpper()))
                    {
                        keysToDelete.Add(keyValue);
                    }
                }
                tgtReader.Close();

                if (keysToDelete.Count > 0)
                {
                    int recordsDeleted = 0;
                    foreach (string keyValueToDelete in keysToDelete)
                    {
                        string keyValExpression = FieldDataType.ToMssqlDataValue(keyFieldDataType, keyValueToDelete);
                        if (this._Instruction.TgtDbProvider == DataProviderType.Oracle)
                        {
                            keyValExpression = FieldDataType.ToOracleDataValue(keyFieldDataType, keyValueToDelete);
                        }
                        tgtCmd.CommandText = "delete " + this._Instruction.TgtTableName + " where " +
                                             uniqueCompositeKeyExp +
                                             "=" + keyValExpression;
                        tgtCmd.CommandType = CommandType.Text;
                        tgtCmd.ExecuteNonQuery();
                        recordsDeleted++;
                        if(this.WorkerStatusChangedEvent !=null && recordsDeleted % 10 ==0)
                        {
                            string msg = string.Format("Deleting {0} of {1} records from table {2}", 
                                recordsDeleted, keysToDelete, this._Instruction.TgtTableName);
                            int percent = recordsDeleted*100/keysToDelete.Count;
                            this.WorkerStatusChangedEvent(this._TicketID, this.GetType(), this._Instruction,
                                this.WorkflowName, this.ComponentName, msg,
                                                          percent);
                        }
                    }
                }

                if (identityInsert)
                {
                    tgtCmd.CommandType = CommandType.Text;
                    tgtCmd.CommandText = string.Format("SET IDENTITY_INSERT {0} OFF;", this._Instruction.TgtTableName);
                    tgtCmd.ExecuteNonQuery();
                }
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                srcConn.Close();
                tgtConn.Close();
            }

            return recordsInserted;
        }

        private DataTable TestTableSyncUsingUniqueKeys()
        {
            DataTable tblTest = new DataTable();
            // DateTime currentTimestamp = DateTime.Now;

            IDbConnection srcConn;
            if (this._Instruction.SrcDbProvider == DataProviderType.Oracle)
            {
                srcConn = new OracleConnection(this._Instruction.SrcConnStr);
            }
            else
            {
                srcConn = new SqlConnection(this._Instruction.SrcConnStr);
            }
            try
            {
                TableColumn[] srcCols = null;
                if (this._Instruction.UseSql)
                {
                    srcCols =
                        DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                               this._Instruction.SelectStatement);
                }
                else
                {
                    srcCols = DbUtil.GetTableColumns(this._Instruction.SrcDbProvider, this._Instruction.SrcConnStr,
                                                     this._Instruction.SrcSchema, this._Instruction.SrcTblName);
                }

                foreach (TableColumn col in srcCols)
                {
                    tblTest.Columns.Add(col.ColumnName, FieldDataType.ToSystemType(col.DbType));
                }

                srcConn.Open();
                string uniqueCompositeKeyName = string.Empty;
                string uniqueCompositeKeyExp = string.Empty;
                for (int i = 0; i < this._Instruction.SrcTableUniqueKeys.Count; i++)
                {
                    uniqueCompositeKeyName += this._Instruction.SrcTableUniqueKeys[i];
                    if (i < this._Instruction.SrcTableUniqueKeys.Count - 1)
                    {
                        uniqueCompositeKeyName += "_";
                    }
                    uniqueCompositeKeyExp += this._Instruction.SrcTableUniqueKeys[i];
                    if (i < this._Instruction.SrcTableUniqueKeys.Count - 1)
                    {
                        uniqueCompositeKeyExp += "+'_'+";
                    }
                }
                IDbCommand srcCmd = srcConn.CreateCommand();
                if (this._Instruction.UseSql)
                {
                    string selectSql =
                        this._Instruction.SelectStatement.Substring(0,
                                                                    this._Instruction.SelectStatement.ToLower().IndexOf(
                                                                        "from"));
                    string fromSql =
                        this._Instruction.SelectStatement.Substring(
                            this._Instruction.SelectStatement.ToLower().IndexOf("from"));

                    if (this._Instruction.SrcTableUniqueKeys.Count > 1)
                    {
                        selectSql += "," + uniqueCompositeKeyExp + " as " + uniqueCompositeKeyName;
                    }
                    selectSql += fromSql +
                                 " order by ";
                    if (this._Instruction.SrcTableUniqueKeys.Count > 1)
                    {
                        selectSql += uniqueCompositeKeyName;
                    }
                    else
                    {
                        selectSql += this._Instruction.SrcTableUniqueKeys[0];
                    }
                    srcCmd.CommandText = selectSql;
                }
                else
                {
                    string selectSql = "select *";
                    if (this._Instruction.SrcTableUniqueKeys.Count > 1)
                    {
                        selectSql += "," + uniqueCompositeKeyExp + " as " + uniqueCompositeKeyName;
                    }
                    selectSql += " from " +
                                 (string.IsNullOrEmpty(this._Instruction.SrcSchema)
                                      ? QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider)
                                      : this._Instruction.SrcSchema + "." + QuoteString1(this._Instruction.SrcDbProvider) + this._Instruction.SrcTblName + QuoteString2(this._Instruction.SrcDbProvider)) +
                                 " order by ";
                    if (this._Instruction.SrcTableUniqueKeys.Count > 1)
                    {
                        selectSql += uniqueCompositeKeyName;
                    }
                    else
                    {
                        selectSql += this._Instruction.SrcTableUniqueKeys[0];
                    }
                    srcCmd.CommandText = selectSql;
                }

                srcCmd.CommandType = CommandType.Text;
                IDataReader srcReader = srcCmd.ExecuteReader();
                int recordTested = 0;
                while (srcReader.Read())
                {
                    DataRow drTest = tblTest.NewRow();
                    foreach (TableColumn srcCol in srcCols)
                    {
                        object srcFieldValue = srcReader[srcCol.ColumnName];
                        drTest[srcCol.ColumnName] = srcFieldValue;
                    }
                    tblTest.Rows.Add(drTest);

                    recordTested++;
                    if (recordTested > 5)
                        break;
                }
                srcReader.Close();
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                srcConn.Close();
            }

            return tblTest;
        }

        private string GetConvertionSqlExpression(DataProviderType dbProvider, string fieldName)
        {
            if(dbProvider==DataProviderType.Oracle)
            {
                return string.Format("TO_CHAR({0})", fieldName);
            }
            else
            {
                return string.Format("cast({0} as varchar)", fieldName);
            }
        }
        #endregion
    }
}
